package com.steakliu.sun.codec;

import com.steakliu.sun.common.thread.FlowControlPostProcessorThreadPool;
import com.steakliu.sun.common.util.CodecUtil;
import com.steakliu.sun.flow.api.FlowControlPostProcessor;
import com.steakliu.sun.protocol.MsgHeader;
import com.steakliu.sun.protocol.SunProtocol;
import com.steakliu.sun.serializer.api.SunSerializer;
import com.steakliu.sun.spi.loader.ExtensionLoader;
import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.MessageToByteEncoder;

import java.nio.charset.StandardCharsets;

/**
 * 功能说明： 编码器
 * <p>
 * Original @Author: steakliu-刘牌, 2023-01-08  12:41
 * <p>
 * Copyright (C)2020-2022  steakliu All rights reserved.
 */
public class SunEncoder extends MessageToByteEncoder<SunProtocol<Object>> {

    private FlowControlPostProcessor flowControlPostProcessor;

    private final boolean isUseFlowControl;

    public SunEncoder(boolean isUseFlowControl, String flowControlType){
        this.isUseFlowControl = isUseFlowControl;
        if (isUseFlowControl){
            this.flowControlPostProcessor = ExtensionLoader.getExtension(FlowControlPostProcessor.class, flowControlType);
        }
    }

    @Override
    protected void encode(ChannelHandlerContext context, SunProtocol<Object> protocol, ByteBuf byteBuf) throws Exception {
        MsgHeader header = protocol.getMsgHeader();
        byteBuf.writeShort(header.getMagic());
        byteBuf.writeByte(header.getType());
        byteBuf.writeByte(header.getStatus());
        byteBuf.writeLong(header.getRequestId());
        //序列化
        String serializerName = header.getSerializer();
        byteBuf.writeBytes(CodecUtil.fillZero(serializerName).getBytes(StandardCharsets.UTF_8));
        SunSerializer serializer = ExtensionLoader.getExtension(SunSerializer.class, serializerName);
        byte[] messageBodyBytes = serializer.serialize(protocol.getBody());

        int msgLen = messageBodyBytes.length;
        byteBuf.writeInt(msgLen);
        byteBuf.writeBytes(messageBodyBytes);
        header.setMsgLen(msgLen);

        //流控切入
        if (this.isUseFlowControl){
            FlowControlPostProcessorThreadPool.submit(() -> {
                flowControlPostProcessor.postProcess(header);
            });
        }
    }
}
